package com.ustadzmenjawab.twitter;

import java.util.Date;
import java.util.Hashtable;
import java.util.concurrent.ConcurrentLinkedQueue;
import javax.naming.SizeLimitExceededException;

import com.ustadzmenjawab.conf.Configuration;
import com.ustadzmenjawab.db.DBAccess;

import twitter4j.FilterQuery;
import twitter4j.Status;
import twitter4j.StatusListener;
import twitter4j.TwitterStream;
import twitter4j.TwitterStreamFactory;
import twitter4j.conf.ConfigurationBuilder;

public class TwitterStreamReader {
	
	public final static int NUM_CONSUMER_THREAD = 30;
	public final static int SLEEP_PERIOD_SEC = 5;
	public static void main(String[] args){
		
		DBAccess db = new DBAccess();
		Hashtable<Long,String> ustadzList = db.getUstadzList();
		
		ConfigurationBuilder cb = new ConfigurationBuilder();
		cb.setDebugEnabled(true)
		.setIncludeRTsEnabled(true)
		.setIncludeEntitiesEnabled(true)
		.setOAuthConsumerKey(Configuration.getProperty("OAUTH_CONSUMER_KEY"))
		.setOAuthConsumerSecret(Configuration.getProperty("OAUTH_CONSUMER_SECRET"))
		.setOAuthAccessToken(Configuration.getProperty("OAUTH_ACCESS_TOKEN"))
		.setOAuthAccessTokenSecret(Configuration.getProperty("OAUTH_ACCESS_TOKEN_SECRET"));
		twitter4j.conf.Configuration twconf = cb.build();
		final ConcurrentLinkedQueue<Status> statusBuffer = new ConcurrentLinkedQueue<Status>();
		TwitterStream twitterStream = new TwitterStreamFactory(twconf).getInstance();
		StatusListener listener = new TwitterStatusListener(statusBuffer);
		twitterStream.addListener(listener);
		
		long[] trackUstadz = new long[ustadzList.size()];
		String[] trackUstadzMention = new String[ustadzList.size()];
		int i = 0;
		for(long userid : ustadzList.keySet()){
			trackUstadz[i] = userid;
			trackUstadzMention[i] = "@"+ustadzList.get(userid);
			i++;
		}
		for(int tid = 0; tid < NUM_CONSUMER_THREAD;tid++){
			runConsumers(statusBuffer,tid);
		}
		
		FilterQuery fq = new FilterQuery(trackUstadz);
		fq.track(trackUstadzMention);
	
		twitterStream.filter(fq);
		//System.out.println("Disconnected from Twitter");
		
	}
	
	public static void runConsumers(final ConcurrentLinkedQueue<Status> statusBuffer, final int threadID ){
		if(statusBuffer == null) 
			return;
		Thread consumerThread = new Thread(new Runnable(){

			@Override
			public void run() {
				System.out.println(new Date() + " Thread ID " + threadID + " started.");
				DBAccess db = new DBAccess();
				while(true){
					Status status = statusBuffer.poll();
					if(status == null){
						System.out.println(new Date() + " Thread ID " + threadID + " goes to sleep.");
						try {
							Thread.sleep(SLEEP_PERIOD_SEC * 1000);
						} catch (InterruptedException e) {
							
							e.printStackTrace();
						}
						continue;
					}
					if(status.isRetweet() ){
						if(!db.isUstadz(status.getUser().getId()) && status.getRetweetedStatus() != null){
							db.insertRetweet(status);
							db.updateReTweetCountSummary(status.getRetweetedStatus().getId(), status.getRetweetedStatus());
							continue;
						}
					}
					
					long convid = db.insertPublicTweet(status);
					if(!status.isRetweet() && status.getInReplyToStatusId() == -1)
						db.insertConvTweetSummary(convid, status);
					db.insertEntities(status);
					db.insertRetweet(status);

					db.insertUser(status.getUser());
					System.out.println(new Date() + " Status ID " + threadID + " inserted.");
				}
			}
			
		});
		consumerThread.start();
	}
}
